package com.fuhao.ack.comfirm;

import com.fuhao.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * ack消息确认comfirm模式之异步模式
 * @author zhang
 */
public class Product3 {
    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare("ack_comfirm_async_queue",true,false,false,null);
        channel.confirmSelect();
        final SortedSet<Long> confirmSet =  Collections.synchronizedSortedSet(new  TreeSet<Long>());

            channel.addConfirmListener(new ConfirmListener() {
                //multiple表示是否是多条，deliveryTag回应tag
                //handAck表示发送成功，则删除集合中对于的数据
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("消息发送成功,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");
                        if(multiple){//接收多条tag
                            confirmSet.headSet(deliveryTag + 1).clear();
                        }else {//一条
                            confirmSet.remove(deliveryTag);
                        }
                }

                /**
                 * 处理返回失败
                 */
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("失败,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");
                    if (multiple) {
                        confirmSet.headSet(deliveryTag + 1).clear();
                    }
                    else {
                        confirmSet.remove(deliveryTag);
                    }
                }
            });

        //发送对条数据
        for(int i = 0;i < 10;i++){
            String msg = "我是confirm模式 消息 异步【" + i + "】";
            long tag = channel.getNextPublishSeqNo();
            //发送消息
            channel.basicPublish("", "ack_comfirm_async_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
            System.out.println("tag:" + tag);
            confirmSet.add(tag);
        }

    }
}
